package com.xxxx

import org.apache.flink.api.common.functions.FlatMapFunction
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.util.Collector

/**
 * @program: flink19Test
 * @author: CoreDao
 * @create: 2021-03-31 20:40
 * */

object StreamingJobWc {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration())
    val socket = env.socketTextStream("localhost", 9999)

    import org.apache.flink.api.scala._
    socket.flatMap(_.split(" ")).map((_,1)).keyBy(_._1).sum(1).print()

    socket.flatMap(new FlatMapFunction[String,(String,Int)] {
      override def flatMap(value: String, out: Collector[(String, Int)]): Unit = {
        val strings = value.split(" ")
        for (s <- strings){
          out.collect((s,1))
        }
      }
    }).keyBy(_._1).sum(1).print()


    env.execute()
  }

}
